/**
 * 
 */
package com.ccb.channel.p8;

import java.io.IOException;
import java.util.Set;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.bson.BasicBSONObject;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ccb.util.Record;
import com.ccb.util.RecordType;
import com.ccb.util.StandardRecord;
import com.mongodb.hadoop.io.BSONWritable;

//import com.alibaba.fastjson.JSON;

/**
 * @author congyang
 *
 */
public class P8ReducerMongo extends Reducer<Text, Text, Text, BSONWritable> {
	String reqXml = null;
	String resXml = null;
	String trace_id = null;
	String service_code = null;
	String[] keys = null;
	String resJson = null;
	String reqJson = null;
	Text output = new Text();

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		StandardRecord standardRecord = new StandardRecord();
		for (Text val : values) {
			Record rd = JSON.parseObject(val.toString(), Record.class);
			standardRecord.setUid(rd.getUid());
			standardRecord.setServiceId(rd.getServiceId());
			if (rd.getRecordType().equals(RecordType.REQUEST)) {
				standardRecord.setRequestData(rd.getData());
				standardRecord.setRequest(rd.getOriginalData());
				standardRecord.setRecordType(RecordType.REQUEST);
				standardRecord.setTime(rd.getTime());

			}
			if (rd.getRecordType().equals(RecordType.RESPONSE)) {
				standardRecord.setResponseData(rd.getData());
				standardRecord.setResponse(rd.getOriginalData());
				standardRecord.setRecordType(RecordType.RESPONSE);
				standardRecord.setResponseCode(rd.getResponseCode());
			}
		}
		if (standardRecord.getRequest() != null && standardRecord.getResponse() != null)
			standardRecord.setRecordType(RecordType.BOTH);
		BasicBSONObject bsonObject = new BasicBSONObject();
		JSONObject jsonObject = (JSONObject) JSON.toJSON(standardRecord);
		Set<String> keySet = jsonObject.keySet();
		for (String string : keySet) {
			bsonObject.append(string, jsonObject.get(string));
		}
		context.write(key, new BSONWritable(bsonObject));
	}

}
